package com.szq.lottery.application.mq;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @author 吾空白
 * @description: 消息消费者
 */
@Component
public class KafkaConsumer {
    private Logger logger= LoggerFactory.getLogger(KafkaConsumer.class);

    @KafkaListener(topics = KafkaProducer2.TOPIC_TEST,groupId = KafkaProducer2.TOPIC_GROUP)
    public void topicTest(ConsumerRecord<?,?> record, Acknowledgment ack,@Header(KafkaHeaders.RECEIVED_TOPIC) String topic){
        Optional<?> message = Optional.ofNullable(record.value());
        if (message.isPresent()){
            Object msg = message.get();
            logger.info("topic_test 消费了：Topic:"+topic+",Message:"+msg);
            ack.acknowledge();
        }
    }
}
